import mrjob
from mrjob.job import MRJob
from mrjob.protocol import JSONProtocol
from hdfs import client
import re


def parse_file(content):
    section_head = None
    section_body = None

    # with open(path, "r", encoding="utf-8") as fp:
    for line in re.split("[\r\n]+", content):
        if not line:
            break
        head = re.match("^section:\\s*(?P<section>\\w+)", line)
        if head:
            if section_head:
                yield section_head, section_body

            section_head = head.group("section")
            section_body = {}
            continue

        body = re.match("(?P<key>\\w+)\\s*=\\s*(?P<value>.+)", line)
        if body:
            if section_head:
                section_body[body.group("key")] = body.group("value")
            continue

            # raise Exception("Can not recognize line: " + line)

    if section_head:
        yield section_head, section_body


# for key, value in parse_file("./logs.txt"):
#     print(key, value)


class MyJob(MRJob):
    def mapper_raw(self, path, uri):
        c = client.Client("http://10.2.2.127:50070")
        with c.read('/logs.txt') as reader:
            for key, value in parse_file(reader.read().decode("utf-8")):
                yield key, value

    def reducer(self, key, value):
        values = list(value)
        yield key, values[0]

    def load_args(self, args):
        super().load_args(args)
        pass    

if __name__ == "__main__":
    job = MyJob(args=["--jobconf", "line=this is a test"])
    job.run()
